package cgl.sensorgrid.sopac.gps.filters;

import java.util.LinkedList;
import java.util.Calendar;
import java.util.Properties;
import java.util.Collections;
import java.util.List;
import java.util.StringTokenizer;
import java.text.SimpleDateFormat;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.io.IOException;
import java.lang.InterruptedException;

import cgl.sensorgrid.filters.Filter;
import cgl.sensorgrid.common.ITRFtoLatLon;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.util.ByteUtilities;
import cgl.narada.service.time.ntptime.TimeServiceImpl;
import cgl.narada.service.time.TimeService;
import cgl.narada.service.*;
import cgl.narada.service.client.*;

import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.DatasetFactory;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.DeploymentException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceException;
import cgl.granules.operation.*;
import cgl.granules.application.*;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.granules.results.ResultsFactory;

/***
 * 
 * This is the filter converting RYO data to ASCII data. It publishes ASCII data
 * results to AsciiMap. This filter uses the methods in class ryoTOasciiFilter.
 * 
 * @author Chao Sun
 * 
 */

class toAsciiFilter {

	private String header = "";
	private String position = "";
	private String xyzVariance = "";
	private String troposphericCovariance = "";
	private String satelliteInfo = "";
	private String RYOmessage = "";
	private String stamp;
	int[] a = new int[1];
	private String asciiIden = "ASCII-Data-Topic";
	private EventProducer producer;
	private ClientService clientService;
	private ResultPayload resultPayload;

	public toAsciiFilter(Properties props, String commType)
			throws ServiceException {
		resultPayload = ResultPayload.getInstance();
		clientService = SessionService.getClientService(1);
		clientService.initializeBrokerCommunications(props, commType);
		initializeProducer();
	}

	private void initializeProducer() throws ServiceException {
		producer = clientService.createEventProducer();
		producer.generateEventIdentifier(false);
		producer.setTemplateId(12345);
		producer.setDisableTimestamp(true);
		producer.setSuppressRedistributionToSource(true);
	}

	public String run(NBEvent nbevent) { // void
		String data = null;
		try {
			byte[] data2 = nbevent.getContentPayload();
			if (data2 != null && data2.length > 0) {
				byte[] processed = processRawMessage(data2);
				if (processed != null) {
					data = new String(processed);
				}
			}
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		
		return data;
	}

	public byte[] processRawMessage(byte[] input) { // void
		byte[] data = null;
		try {
			boolean validHeader = false;
			if (input.length >= 5) {
				validHeader = decodeHeader(input);
			}
			if (validHeader && input.length > 52) {
				decodePosition(input, 5);
			}

			String pos = this.getPosition();
			if (pos.endsWith("\n")) {
				pos = pos.substring(0, pos.length() - 1);
			}
			if (pos != null && pos.trim() != "" && pos.trim().length() > 0) {
				// publishData((pos + "\nEOF\n").getBytes());
				data = (pos + "\nEOF\n").getBytes();
			}
			this.position = "";
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		
		return data;
	}

	public void publishData(String data) throws DatasetException,
			ServiceException, MapReduceException, MarshallingException,
			InterruptedException {
		NBEvent nbEvent = producer.generateEvent(
				TemplateProfileAndSynopsisTypes.STRING, asciiIden,
				resultPayload.getBytes(data));
		producer.publishEvent(nbEvent);
	}

	public String messageToGML(String message) {
		try {
			return "";
		} catch (Exception ex) {
			ex.printStackTrace();
			return null;
		}
	}

	public boolean decodeHeader(byte[] input) {
		try {
			boolean validHeader = false;
			cgl.narada.util.ByteUtilities b = new cgl.narada.util.ByteUtilities(
					a);
			int SyncMarker1 = b.getIntFromByte(input[0]);
			int SyncMarker2 = b.getIntFromByte(input[1]);
			if (Integer.toHexString(SyncMarker1).toUpperCase()
					.equalsIgnoreCase("9C")
					&& Integer.toHexString(SyncMarker2).toUpperCase().equals(
							"A5")) {
				validHeader = true;
			}
			return validHeader;
		} catch (Exception ex) {
			ex.printStackTrace();
			return false;
		}
	}

	public String decodePosition(byte[] input, int offset) {
		try {
			String line = "";
			ByteUtilities b = new ByteUtilities(a);
			int i = offset;
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);

			// Date
			short GPSWeek = bb.getShort(i);
			int GPSmsOfWeek = bb.getInt(i + 2);
			String date = getDate(Short.toString(GPSWeek), Integer
					.toString(GPSmsOfWeek));
			// Site info
			int SiteIndex = b.getIntFromByte(input[i + 6]);
			int SiteCount = b.getIntFromByte(input[i + 7]);
			String SiteId = Character.toString((char) (input[i + 8]))
					+ Character.toString((char) (input[i + 9]))
					+ Character.toString((char) (input[i + 10]))
					+ Character.toString((char) (input[i + 11]));

			// Skip 12 and 13
			line += "\t" + date;
			line += "\t" + SiteId;
			line += "\t" + SiteIndex;
			line += "\t" + SiteCount;

			// ITRF Positions
			double X = bb.getDouble((i + 14));
			double Y = bb.getDouble((i + 22));
			double Z = bb.getDouble((i + 30));
			double T = bb.getDouble((i + 38));
			line += "\t" + X;
			line += "\t" + Y;
			line += "\t" + Z;
			line += "\t" + T;

			// Lat, Lon, Height positions
			double lat = 0.0;
			double lon = 0.0;
			double hi = 0.0;

			ITRFtoLatLon latlon = new ITRFtoLatLon();
			double[] llh = null;
			llh = latlon.GetLatLonHeightRadians(X, Y, Z);

			if (llh != null && llh.length == 3) {
				line += "\t" + llh[0];
				line += "\t" + llh[1];
				line += "\t" + llh[2];
			}

			String posQualityBytes = b.printByte(input[i + 46]);
			if (!posQualityBytes.equals("") && posQualityBytes.length() == 8) {
				String method = posQualityBytes.substring(3, 6);
				String calc = posQualityBytes.substring(6, 8);

				if (calc.equals("00")) {
					line += " Undetermined";
				} else if (calc.equals("01")) {
					line += " L1";
				} else if (calc.equals("10")) {
					line += " L1/L2";
				}

				if (method.equals("001")) {
					line += " Undetermined";
				} else if (method.equals("010")) {
					line += " Absolute";
				} else if (method.equals("011")) {
					line += " Relative";
				} else if (method.equals("100")) {
					line += " Phase";
				}
			}

			String flagString = b.printByte(input[i + 47]);
			boolean xyz = false;
			boolean tropo = false;
			boolean satellite = false;

			if (!flagString.equals("") && flagString.length() == 8) {
				if (flagString.charAt(7) == '1') {
					line += " XYZ";
					xyz = true;
				}
				if (flagString.charAt(6) == '1') {
					line += " Troposhpheric";
					tropo = true;
				}
				if (flagString.charAt(5) == '1') {
					line += " Satellite";
					satellite = true;
				}
			}

			i += 47;

			if (input.length > 51) {
				if (xyz == true) {
					i = decodeXYZVariances(input, i + 1);
				}
				line += " " + this.getXyzVariance();
				this.setXyzVariance("");
			}

			if (input.length > i) {
				if (satellite == true) {
					i = decodeSatellite(input, i + 1);
				}
				line += " " + this.getSatelliteInfo();
				this.setSatelliteInfo("");
			}

			if (line != null && !line.equals("")) {
				this.appendPosition(line.trim() + "\n");
			}

			if (input.length > i + 52) {
				decodePosition(input, i + 7);
			}

			return line;
		} catch (Exception ex) {
			System.err.println("\nException occured in decodePosition()");
			ex.printStackTrace();
			return null;
		}
	}

	public int decodeXYZVariances(byte[] input, int i) {
		try {
			String line = "";
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);

			double scale = bb.getDouble(i);
			double Xvar = bb.getDouble((i + 8));
			double Yvar = bb.getDouble((i + 16));
			double Zvar = bb.getDouble((i + 24));
			double YXvar = bb.getDouble((i + 32));
			double YZvar = bb.getDouble((i + 40));
			double ZXvar = bb.getDouble((i + 48));

			i += 55;
			line += " " + scale;
			line += " " + Xvar;
			line += " " + Yvar;
			line += " " + Zvar;
			line += " " + YXvar;
			line += " " + YZvar;
			line += " " + ZXvar;
			this.setXyzVariance(line);
			return i;
		} catch (Exception ex) {
			return i;
		}

	}

	public int decodeTroposphericCoverianceBlock(byte[] input, int i) {
		try {
			String line = "";
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);

			double Tvar = bb.getDouble((i + 8));
			double TXcovar = bb.getDouble((i + 16));
			double TYcovar = bb.getDouble((i + 24));
			double TZcovar = bb.getDouble((i + 32));

			i += 33;
			line += " " + Tvar;
			line += " " + TXcovar;
			line += " " + TYcovar;
			line += " " + TZcovar;
			this.setTroposphericCovariance(line);
			return i;
		} catch (Exception ex) {
			return i;
		}

	}

	public int decodeSatellite(byte[] input, int i) {
		try {
			String line = "";
			ByteUtilities b = new ByteUtilities(a);
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);

			int satelliteCount = b.getIntFromByte(input[i]);
			double GDOP = bb.getDouble((i + 1));

			if (satelliteCount > 0) {
				i += 9;
				for (int j = 0; j < satelliteCount; j++) {
					if (input.length > i + 5) {
						int PRN = b.getIntFromByte(input[i]);
						int PRNFlags = b.getIntFromByte(input[i + 1]);
						short Elevation = bb.getShort(i + 2);
						short Azimuth = bb.getShort(i + 4);
						line += " " + (j + 1);
						line += " " + PRN;
						line += " " + PRNFlags;
						line += " " + Elevation;
						line += " " + Azimuth;
						i += 6;
					}
				}
				this.setSatelliteInfo(line);
			}
			return i;
		} catch (Exception ex) {
			return i;
		}
	}

	public void setHeader(String header) {
		this.header = header;
	}

	public void setPosition(String position) {
		this.position = position;
	}

	public void appendPosition(String pos) {
		this.position += pos;
	}

	public void setXyzVariance(String xyzVariance) {
		this.xyzVariance = xyzVariance;
	}

	public void setTroposphericCovariance(String troposphericCovariance) {
		this.troposphericCovariance = troposphericCovariance;
	}

	public void setSatelliteInfo(String satelliteInfo) {
		this.satelliteInfo = satelliteInfo;
	}

	public void setRYOmessage(String header, String position) {
		String RYOmsg = header;
		RYOmsg.trim();

		RYOmsg += position;
		RYOmsg.trim();
		this.RYOmessage = RYOmsg;
	}

	public String getHeader() {
		return header;
	}

	public String getPosition() {
		return position;
	}

	public String getXyzVariance() {
		return xyzVariance;
	}

	public String getTroposphericCovariance() {
		return troposphericCovariance;
	}

	public String getSatelliteInfo() {
		return satelliteInfo;
	}

	public String getRYOmessage() {
		return RYOmessage;
	}

	public String getDate(String GPS_week, String GPS_millisecond) {
		try {
			long GPS_start_epoch = 315964800;
			long GPS_leapseconds = 13;

			long time = GPS_start_epoch + Long.parseLong(GPS_week) * 7 * 24
					* 3600 + Long.parseLong(GPS_millisecond) / 1000
					- GPS_leapseconds;

			Calendar cal = Calendar.getInstance();
			cal.setTimeInMillis(time * 1000);
			SimpleDateFormat myDateFormat = new SimpleDateFormat(
					"yyyy-MM-dd\thh:mm:ssa-z");

			return myDateFormat.format(cal.getTime());
		} catch (Exception ex) {
			ex.printStackTrace();
			return null;
		}
	}

}

/***
 * 
 * This is the filter converting RYO data to POS data. It publishes POS data
 * results to PosMap. This filter uses the methods in class ryoTOposFilter.
 * 
 * @author Chao Sun
 * 
 */

class toPosFilter {

	private String header = "";
	private String position = "";
	private String xyzVariance = "";
	private String troposphericCovariance = "";
	private String satelliteInfo = "";
	private String RYOmessage = "";
	int[] a = new int[1];
	private String posIden = "POS-Data-Topic";
	private EventProducer producer;
	private ClientService clientService;
	private ResultPayload resultPayload;

	public toPosFilter(Properties props, String commType)
			throws ServiceException {
		resultPayload = ResultPayload.getInstance();
		clientService = SessionService.getClientService(2);
		clientService.initializeBrokerCommunications(props, commType);
		initializeProducer();
	}

	private void initializeProducer() throws ServiceException {
		producer = clientService.createEventProducer();
		producer.generateEventIdentifier(false);
		producer.setTemplateId(54321);
		producer.setDisableTimestamp(true);
		producer.setSuppressRedistributionToSource(true);
	}

	public String run(NBEvent nbevent) { // void
		String data = null;
		try {
			byte[] data2 = nbevent.getContentPayload();
			if (data2 != null && data2.length > 0) {
				byte[] processed = processRawMessage(data2);
				if (processed != null) {
					data = new String(processed);
				}
			}
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		
		return data;
	}

	public byte[] processRawMessage(byte[] input) { // void
		byte[] data = null;
		try {
			boolean validHeader = false;
			if (input.length >= 5) {
				validHeader = decodeHeader(input);
			}
			if (validHeader && input.length > 52) {
				decodePosition(input, 5);
			}

			String pos = this.getPosition();
			if (pos.endsWith("\n")) {
				pos = pos.substring(0, pos.length() - 1);
			}
			if (pos != null && pos.trim() != "" && pos.trim().length() > 0) {
				// publishData((pos + "\nEOF\n").getBytes());
				data = (pos + "\nEOF\n").getBytes();
			}
			this.position = "";
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		
		return data;
	}

	public void publishData(String data) throws DatasetException,
			ServiceException, MapReduceException, MarshallingException,
			InterruptedException {
		NBEvent nbEvent = producer.generateEvent(
				TemplateProfileAndSynopsisTypes.STRING, posIden, resultPayload
						.getBytes(data));
		producer.publishEvent(nbEvent);
	}

	public String messageToGML(String message) {
		try {
			return "";
		} catch (Exception ex) {
			ex.printStackTrace();
			return null;
		}
	}

	public boolean decodeHeader(byte[] input) {
		try {
			boolean validHeader = false;
			cgl.narada.util.ByteUtilities b = new cgl.narada.util.ByteUtilities(
					a);

			int SyncMarker1 = b.getIntFromByte(input[0]);
			int SyncMarker2 = b.getIntFromByte(input[1]);
			if (Integer.toHexString(SyncMarker1).toUpperCase()
					.equalsIgnoreCase("9C")
					&& Integer.toHexString(SyncMarker2).toUpperCase().equals(
							"A5")) {
				validHeader = true;
			}
			return validHeader;
		} catch (Exception ex) {
			ex.printStackTrace();
			return false;
		}
	}

	public String decodePosition(byte[] input, int offset) {
		try {
			String line = "";
			ByteUtilities b = new ByteUtilities(a);
			int i = offset;
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);

			// Date
			short GPSWeek = bb.getShort(i);
			int GPSmsOfWeek = bb.getInt(i + 2);
			String date = getDate(Short.toString(GPSWeek), Integer
					.toString(GPSmsOfWeek));

			// Site info
			int SiteIndex = b.getIntFromByte(input[i + 6]);
			int SiteCount = b.getIntFromByte(input[i + 7]);
			String SiteId = Character.toString((char) (input[i + 8]))
					+ Character.toString((char) (input[i + 9]))
					+ Character.toString((char) (input[i + 10]))
					+ Character.toString((char) (input[i + 11]));

			// Skip 12 and 13
			line += "\t" + SiteId;
			line += "\t" + date;

			// ITRF Positions
			double X = bb.getDouble((i + 14));
			double Y = bb.getDouble((i + 22));
			double Z = bb.getDouble((i + 30));
			double T = bb.getDouble((i + 38));
			line += "\t" + X;
			line += "\t" + Y;
			line += "\t" + Z;
			line += "\t" + T;

			// Lat, Lon, Height positions
			double lat = 0.0;
			double lon = 0.0;
			double hi = 0.0;
			ITRFtoLatLon latlon = new ITRFtoLatLon();
			double[] llh = null;
			llh = latlon.GetLatLonHeightRadians(X, Y, Z);

			if (llh != null && llh.length == 3) {
				line += "\t" + llh[0];
				line += "\t" + llh[1];
				line += "\t" + llh[2];
			}

			if (input.length > i + 47) {
				String flagString = b.printByte(input[i + 47]);
				boolean xyz = false;
				boolean tropo = false;
				boolean satellite = false;

				if (!flagString.equals("") && flagString.length() == 8) {
					if (flagString.charAt(7) == '1') {
						xyz = true;
					}
					if (flagString.charAt(6) == '1') {
						tropo = true;
					}
					if (flagString.charAt(5) == '1') {
						satellite = true;
					}
				}

				i += 47;

				if (input.length > 51) {
					if (xyz == true) {
						i = decodeXYZVariances(input, i + 1);
					}
					line += " " + this.getXyzVariance();
					this.setXyzVariance("");
				}

				if (input.length > i) {
					if (satellite == true) {
						i = decodeSatellite(input, i + 1);
					}
					line += " " + this.getSatelliteInfo();
					this.setSatelliteInfo("");
				}

				if (line != null && !line.equals("")) {
					this.appendPosition(line.trim() + "\n");
				}

				if (input.length > i + 52) {
					decodePosition(input, i + 7);
				}
			}
			return line;
		} catch (Exception ex) {
			System.err.println("\nException occured in decodePosition()");
			ex.printStackTrace();
			return null;
		}
	}

	public int decodeXYZVariances(byte[] input, int i) {
		try {
			String line = "";
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);
			i += 55;
			return i;
		} catch (Exception ex) {
			return i;
		}

	}

	public int decodeTroposphericCoverianceBlock(byte[] input, int i) {
		try {
			String line = "";
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);
			i += 33;
			this.setTroposphericCovariance(line);
			return i;
		} catch (Exception ex) {
			return i;
		}

	}

	public int decodeSatellite(byte[] input, int i) {
		try {
			String line = "";
			ByteUtilities b = new ByteUtilities(a);
			ByteBuffer bb = ByteBuffer.wrap(input);
			bb.order(ByteOrder.LITTLE_ENDIAN);
			int satelliteCount = b.getIntFromByte(input[i]);
			double GDOP = bb.getDouble((i + 1));

			if (satelliteCount > 0) {
				i += 9;
				for (int j = 0; j < satelliteCount; j++) {
					if (input.length > i + 5) {
						i += 6;
					}
				}
				this.setSatelliteInfo(line);
			}
			return i;
		} catch (Exception ex) {
			return i;
		}
	}

	public void setHeader(String header) {
		this.header = header;
	}

	public void setPosition(String position) {
		this.position = position;
	}

	public void appendPosition(String pos) {
		this.position += pos;
	}

	public void setXyzVariance(String xyzVariance) {
		this.xyzVariance = xyzVariance;
	}

	public void setTroposphericCovariance(String troposphericCovariance) {
		this.troposphericCovariance = troposphericCovariance;
	}

	public void setSatelliteInfo(String satelliteInfo) {
		this.satelliteInfo = satelliteInfo;
	}

	public void setRYOmessage(String header, String position) {
		String RYOmsg = header;
		RYOmsg.trim();
		RYOmsg += position;
		RYOmsg.trim();
		this.RYOmessage = RYOmsg;
	}

	public String getHeader() {
		return header;
	}

	public String getPosition() {
		return position;
	}

	public String getXyzVariance() {
		return xyzVariance;
	}

	public String getTroposphericCovariance() {
		return troposphericCovariance;
	}

	public String getSatelliteInfo() {
		return satelliteInfo;
	}

	public String getRYOmessage() {
		return RYOmessage;
	}

	public String getDate(String GPS_week, String GPS_millisecond) {
		try {
			long GPS_start_epoch = 315964800;
			long GPS_leapseconds = 13;
			long time = GPS_start_epoch + Long.parseLong(GPS_week) * 7 * 24
					* 3600 + Long.parseLong(GPS_millisecond) / 1000
					- GPS_leapseconds;

			Calendar cal = Calendar.getInstance();
			cal.setTimeInMillis(time * 1000);
			SimpleDateFormat myDateFormat = new SimpleDateFormat(
					"yyyy-MM-dd\thh:mm:ssa-z");

			return myDateFormat.format(cal.getTime());
		} catch (Exception ex) {
			ex.printStackTrace();
			return null;
		}
	}
}

/***
 * 
 * This class deploys two filters to convert RYO data to ASCII and POS data.
 * 
 * @author Chao Sun
 * 
 */

class toAsciiPos extends Filter {

	private toAsciiFilter taf = null;
	private toPosFilter tpf = null;
	private AsciiMap asciiMap = null;
	private PosMap posMap = null;
	private RyoReduce reduce = null;
	private String asciiData = null;
	private String posData = null;
	private int lineCount = 0;

	public toAsciiPos(String[] args, String[] filterSpecificArgs) {
		if (args == null || args.length == 0) {
			System.err.println("Incorrect argument number."
					+ "toAsciiPos::Constructor");
			System.exit(0);
		} else {
			initialize(args);
			try {
				initializeReduce(args[4]);
				initializeMaps();
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}
	}

	public void start(Properties props, String commType) {
		try {
			taf = new toAsciiFilter(props, commType);
			tpf = new toPosFilter(props, commType);			
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public void onEvent(NBEvent nbEvent) {
		try {
			if (nbEvent != null && nbEvent.getContentPayload() != null
					&& taf != null && tpf != null) {
				asciiData = taf.run(nbEvent);
				posData = tpf.run(nbEvent);
				nbEvent = null;
				
				if (asciiData != null) {
					taf.publishData(asciiData);
				}
				Thread.sleep(100);
				if (posData != null) {
					tpf.publishData(posData);
				}
				asciiData = null;
				posData = null;
			}
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	private void initializeReduce(String topic) throws DatasetException,
			MapReduceException {
		reduce = new RyoReduce();
		reduce.setAsExecuteWhenDataAvailable();

		Properties props = new Properties();
		props.put("pubTopic", topic);
		reduce.setProcessingDirectives(props);
	}

	private void initializeMaps() throws DatasetException, MapReduceException {
		DatasetFactory datasetFactory = DatasetFactory.getInstance();

		// initialize AsciiMap
		asciiMap = new AsciiMap();
		Properties aprops = new Properties();
		aprops.put("DataType", "ASCII");
		asciiMap.setProcessingDirectives(aprops);

		StreamingAccess aStreamingAccess = datasetFactory
				.createStreamingDataset("ASCII-Data-Map", "RYO-to-ASCII");
		aStreamingAccess.addInputStream("ASCII-Data-Topic",
				StreamingAccess.STRING_SYNOPSIS);
		DatasetCollection aDatasetCollection = asciiMap.getDatasetCollection();
		aDatasetCollection.addDataset(aStreamingAccess);

		asciiMap.setAsExecuteWhenDataAvailable();
		asciiMap.addReduce(reduce);

		// initialize PosMap
		posMap = new PosMap();
		Properties pprops = new Properties();
		pprops.put("DataType", "POS");
		posMap.setProcessingDirectives(pprops);

		StreamingAccess pStreamingAccess = datasetFactory
				.createStreamingDataset("POS-Data-Map", "RYO-to-POS");
		pStreamingAccess.addInputStream("POS-Data-Topic",
				StreamingAccess.STRING_SYNOPSIS);
		DatasetCollection pDatasetCollection = posMap.getDatasetCollection();
		pDatasetCollection.addDataset(pStreamingAccess);

		posMap.setAsExecuteWhenDataAvailable();
		posMap.addReduce(reduce);
	}

	public PosMap getPosMap() {
		return posMap;
	}

	public AsciiMap getAsciiMap() {
		return asciiMap;
	}

	public RyoReduce getReduce() {
		return reduce;
	}

}

/***
 * 
 * This class gets arguments from commandline and use them to generate two
 * filters, which are responsible for converting RYO data to ASCII and POS
 * format respectively. It also deploys the map instances generated by these
 * filters along with a reduce instance binding to the maps.
 * 
 * @author Chao Sun
 * 
 */

class ryoProcessor extends InstanceDeployer {

	private toAsciiPos toAP;
	private ProgressTracker progressTracker;

	public ryoProcessor(String[] args) throws CommunicationsException,
			IOException, MarshallingException, DeploymentException,
			DatasetException, InterruptedException, MapReduceException {
		Properties streamingProperties = new Properties();
		streamingProperties.put("hostname", args[1]);
		streamingProperties.put("portnum", args[2]);
		initialize(streamingProperties);
		toAP = new toAsciiPos(args, null);
	}

	private void run(String[] args) throws CommunicationsException,
			IOException, MarshallingException, DeploymentException,
			DatasetException, InterruptedException, MapReduceException {
		//Thread.sleep(1000);

		RyoReduce reduces[] = new RyoReduce[1];
		reduces[0] = toAP.getReduce();
		Operation maps[] = new Operation[2];
		maps[0] = toAP.getAsciiMap();
		maps[1] = toAP.getPosMap();

		resetTrackers();
		try {
			progressTracker = deployOperations(reduces);
			String jobId = progressTracker.getIdentifier();

			Thread.sleep(500);
			deployOperations(maps, jobId);
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		Properties streamingProperties = new Properties();
		streamingProperties.put("hostname", args[1]);
		streamingProperties.put("portnum", args[2]);
		toAP.start(streamingProperties, "tcp");
	}

	public static void main(String args[]) {
		if (args.length < 5) {
			System.out.println("Usage: java "
					+ "cgl.sensorgrid.sopac.gps.filters.RyoProcessor "
					+ "commType nbHost nbPort subTopic pubTopic");
			System.exit(0);
		}

		String commType = args[0];
		String nbHost = args[1];
		String nbPort = args[2];
		String subTopic = args[3];
		String pubTopic = args[4];

		try {
			String[] argsa = { commType, nbHost, nbPort, subTopic, pubTopic };
			ryoProcessor instance = new ryoProcessor(argsa);
			instance.run(argsa);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}